package com.shengzai.flink.core

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala._

object Demo2WorldCountBatch {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    /**
     * flink处理模式
     * 1、BATCH: 批处理模式，用于处理有界流（hdfs）， 输出最终结果，任务会结束
     * 2、STREAMING（默认）: 流处理模式，用于处理无界流和有界流（kafka,hdfs），输出连续的结果
     *
     *
     * BATCH模式先执行上游task再执行下游task， 在上游同样可以对数据做预聚合
     * STREAMING:模式上游task和下游task同时启动，等待数据到达
     *
     * BATCH模式不能用于处理无界流
     *
     * flink流批统一
     * 思想：再flink中将有界流看作是无界流的一个特例
     * 同一个一套APi既能做流处理也能做批处理
     */
    //设置处理方式
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    val worldDS: DataStream[String] = env.readTextFile("data/worlds.txt")

    val resDS: DataStream[(String, Int)] = worldDS
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)

    resDS.print()

    env.execute()

  }

}
